/*
 * @(#)SynOfflineData.java   1.0  2021年11月17日
 * 
 * Copyright (c)	2014-2020. All Rights Reserved.	GuangZhou hhmk Technology Company LTD.
 */
package com.swift.dts.offline.exec;

import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Callable;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.dao.DuplicateKeyException;
import org.springframework.stereotype.Component;

import com.swift.dao.db.datasource.HandlerDataSource;
import com.swift.dao.db.springjdbc.Jdbc;
import com.swift.dts.dao.OfflineStateInfoDao;
import com.swift.dts.dao.TablesDao;
import com.swift.dts.dao.model.SynObjectKey;
import com.swift.dts.loadconfig.FileTablesConfig;
import com.swift.dts.loadconfig.model.TablesConfig;
import com.swift.dts.model.Table;
import com.swift.dts.util.CreateExecSql;
import com.swift.dts.util.FormatRegexUtil;
import com.swift.exception.extend.SystemException;
import com.swift.util.exec.ThreadUtil;
import com.swift.util.type.TypeUtil;

/**
 * 添加说明
 * 
 * @author zhengjiajin
 * @version 1.0 2021年11月17日
 */
@Component
public class SynOfflineData {

    private static final Logger log = LoggerFactory.getLogger(SynOfflineData.class);

    @Autowired
    public Jdbc jdbc;
    @Autowired
    private FileTablesConfig fileTablesConfig;
    @Autowired
    private OfflineStateInfoDao offlineStateInfoDao;
    // 进度计数器
    private Map<SynObjectKey, AtomicInteger> COUNT_INDEX = new ConcurrentHashMap<>();
    // 停止同步针
    private Map<SynObjectKey, AtomicBoolean> STATE_INDEX = new ConcurrentHashMap<>();

    public void stop(SynObjectKey key) {
        if (STATE_INDEX.containsKey(key)) STATE_INDEX.get(key).set(false);
    }

    public void synOfflineData(SynObjectKey key) {
        // 如对方表内已存在数据,则不同步,异常出错下需要手动清空数据
        if (checkNoSynData(key)) return;
        int countOld = countOldNum(key);
        log.info("离线同步表:" + key.toString() + "需要同步" + countOld);
        if (countOld == 0) return;
        ExecutorService executorService = Executors.newFixedThreadPool(10);
        COUNT_INDEX.put(key, new AtomicInteger());
        STATE_INDEX.put(key, new AtomicBoolean(true));
        int limit = 1000;
        int exeNum = countOld / limit;
        List<Future<Exception>> futureList = new LinkedList<>();
        for (int page = 0; page <= exeNum; page++) {
            Future<Exception> future = executorService.submit(new InsertExeThread(key, page, limit));
            futureList.add(future);
        }
        executorService.shutdown();
        while (!executorService.isTerminated()) {
            updateStatePercent(key, countOld);
            ThreadUtil.sleep(5000);
        }
        updateStatePercent(key, countOld);
        STATE_INDEX.remove(key);
        COUNT_INDEX.remove(key);
        for (Future<Exception> future : futureList) {
            if (future.isDone()) {
                try {
                    Exception ex = future.get();
                    if (ex != null) throw ex;
                } catch (Exception e) {
                    throw new SystemException(e);
                }
            } else {
                throw new SystemException("系统异常,存在线程未完成");
            }
        }
        log.info("离线同步表完成:" + key.toString() + "已同步" + countOld);
    }

    /**
     * @param key
     * @param countOld
     */
    private void updateStatePercent(SynObjectKey key, int countOld) {
        int num = COUNT_INDEX.get(key).get();
        log.info("离线同步表:" + key.toString() + "已经同步" + num + "个,共" + countOld + "个");
        double percent = 0;
        if (countOld == 0) percent = 100d;
        else percent = Double.valueOf(num) / Double.valueOf(countOld) * 100d;
        offlineStateInfoDao.updatePercent(key, num, countOld, percent);
    }

    private class InsertExeThread implements Callable<Exception> {

        private SynObjectKey key;
        // 0开始
        private int page;

        private int limit;

        private InsertExeThread(SynObjectKey key, int page, int limit) {
            this.key = key;
            this.page = page;
            this.limit = limit;
        }

        /**
         * @see java.lang.Runnable#run()
         */
        @Override
        public Exception call() {
            log.info("开始执行" + key + ";" + page);
            AtomicBoolean state = STATE_INDEX.get(key);
            if (state.get() == false) return null;
            try {
                List<Map<String, Object>> list = getDatalist(key, page * limit, limit);
                exceInsert(key, list);
                if (COUNT_INDEX.containsKey(key)) COUNT_INDEX.get(key).addAndGet(list.size());
            } catch (Exception ex) {
                return ex;
            }
            log.info("结束执行" + key + ";" + page);
            return null;
        }

    }

    @SuppressWarnings("unchecked")
    private boolean checkNoSynData(SynObjectKey key) {
        TablesConfig onlyDdlFilter = fileTablesConfig.findConfig(TablesDao.onlyDdlFilter);
        if (onlyDdlFilter != null) {
            boolean blackFilterRegex = FormatRegexUtil.isRegex(new Table(key.getDbName(), key.getTableName()),
                (List<String>) onlyDdlFilter.getRegex());
            if (blackFilterRegex) return true;
        }
        int countNew = countNewNum(key);
        if (countNew > 0) return true;
        return false;
    }

    private List<Map<String, Object>> getDatalist(SynObjectKey key, int i, int limit) {
        String querySql = "SELECT * FROM " + oldDbTable(key) + " LIMIT " + i + "," + limit;
        HandlerDataSource.putDataSource(key.getDataSource());
        List<Map<String, Object>> list = jdbc.queryForList(String.format(querySql, i));
        HandlerDataSource.clear();
        return list;
    }

    // 最大不能超过4M,保险起见不超2M
    private static final int batchByteSize = 1024 * 1024 * 2;

    private void exceInsert(SynObjectKey key, List<Map<String, Object>> list) {
        if (TypeUtil.isNull(list)) return;
        List<String> sqlList = new LinkedList<>();
        for (int i = 0; i < list.size(); i++) {
            Map<String, Object> map = list.get(i);
            String sql = "INSERT INTO " + newDbTable(key) + "(" + keys(map) + ") VALUES(" + values(map) + ")";
            sqlList.add(sql);
        }
        HandlerDataSource.putDataSource(key.getDestination());
        try {
            int byteSize = 0;
            List<String> execSqlList = new LinkedList<>();
            for (String sql : sqlList) {
                byteSize = byteSize + sql.getBytes().length;
                execSqlList.add(sql);
                if (byteSize >= batchByteSize) {
                    jdbc.batchUpdate(execSqlList.toArray(new String[execSqlList.size()]));
                    execSqlList = new LinkedList<>();
                    byteSize = 0;
                }
            }
            jdbc.batchUpdate(execSqlList.toArray(new String[execSqlList.size()]));
        } catch (DuplicateKeyException wx) {
            for (String sql : sqlList) {
                try {
                    jdbc.execute(sql);
                } catch (DuplicateKeyException keyex) {
                }
            }
        }
        HandlerDataSource.clear();
    }

    private String keys(Map<String, Object> map) {
        StringBuffer sql = new StringBuffer();
        for (String column : map.keySet()) {
            sql.append(column + ",");
        }
        sql.delete(sql.length() - 1, sql.length());
        return sql.toString();
    }

    private String values(Map<String, Object> map) {
        StringBuffer sql = new StringBuffer();
        for (String column : map.keySet()) {
            sql.append(CreateExecSql.value(map.get(column)) + ",");
        }
        sql.delete(sql.length() - 1, sql.length());
        return sql.toString();
    }

    private int countNewNum(SynObjectKey key) {
        HandlerDataSource.putDataSource(key.getDestination());
        String countSql = "SELECT COUNT(*) FROM " + newDbTable(key);
        int count = jdbc.queryForInt(countSql);
        HandlerDataSource.clear();
        return count;
    }

    private int countOldNum(SynObjectKey key) {
        HandlerDataSource.putDataSource(key.getDataSource());
        String countSql = "SELECT COUNT(*) FROM " + oldDbTable(key);
        int count = jdbc.queryForInt(countSql);
        HandlerDataSource.clear();
        return count;
    }

    private String newDbTable(SynObjectKey key) {
        return CreateExecSql.dbTable(key.getDbName(), key.getTableName());
    }

    private String oldDbTable(SynObjectKey key) {
        return key.getDbName() + "." + key.getTableName();
    }
}
